package mysql

import (
	"database/sql"
	"errors"
	"fmt"
	"github.com/blastrain/vitess-sqlparser/sqlparser"
	_ "github.com/go-sql-driver/mysql"
	"miniDW/src/datapack"
	"miniDW/src/interfaces"
	"miniDW/src/log"
	"reflect"
	"strings"
	"time"
)

type MysqlSource struct {
	cfg    *MysqlConfig
	db     *sql.DB
	stmt   *sql.Stmt
	offset int
}

func (s *MysqlSource) SourceTableStructSetFinished(ctx interfaces.IContext) error {
	//TODO implement me
	panic("Source do not implement SourceTableStructSetFinished function")
}

func (s *MysqlSource) Process(ctx interfaces.IContext) error {
	return s.ReadRow(ctx)
}

func (s *MysqlSource) buildQuerySql(ts *datapack.TableStruct) error {
	var err error
	fieldNames := ts.GetFeildsNames()
	//fmt.Printf("fieldNames: %v\n", fieldNames)
	if len(fieldNames) == 0 {
		return log.Error(errors.New("field names is empty"))
	}
	query := fmt.Sprintf("SELECT %s FROM %s WHERE 1=1",
		strings.Join(fieldNames, ","),
		ts.Name)
	expression := ts.GetConditionExpression()
	if len(expression) > 0 {
		query += " AND " + expression
	}

	query += " LIMIT ? OFFSET ?"

	s.stmt, err = s.db.Prepare(query)
	if err != nil {
		return log.Error(err)
	}
	return nil
}

func (s *MysqlSource) getSchemaStruct(oldTs *datapack.TableStruct) (*datapack.TableStruct, error) {
	var name string
	var dataType string
	if oldTs.Name() == "" {
		return nil, log.Error(errors.New("table name is empty"))
	}
	queryStr := "SELECT Column_name, Column_type from INFORMATION_SCHEMA.COLUMNS where TABLE_SCHEMA=? AND TABLE_NAME=?"

	args := []any{
		s.cfg.Database,
		oldTs.Name(),
	}

	if oldTs.Length() > 0 {
		fields := oldTs.GetFeildsNames()
		queryStr += " AND COLUMN_NAME IN (" + strings.TrimSuffix(strings.Repeat("?,", len(fields)), ",") + ")"
		for _, field := range fields {
			args = append(args, field)
		}
	}

	newTs := oldTs.Copy()
	newTs.ClearFieldTypes()

	rows, err := s.db.Query(queryStr, args...)
	if err != nil {
		return nil, log.Error(err)
	}
	defer rows.Close()
	for rows.Next() {
		if err := rows.Scan(&name, &dataType); err != nil {
			return nil, log.Error(err)
		}

		field, err := newFieldType(name, dataType)
		if err != nil {
			return nil, err
		}
		err = newTs.AddFieldType(*field)
		if err != nil {
			return nil, err
		}
	}

	err = s.buildQuerySql(newTs)
	if err != nil {
		return nil, log.Error(err)
	}
	return newTs, nil
}

func (s *MysqlSource) getSqlStruct(oldTs *datapack.TableStruct) (*datapack.TableStruct, error) {
	script := oldTs.Sql()
	stmt, err := sqlparser.Parse(script)
	if err != nil {
		return nil, log.Error(err)
	}

	newTs := datapack.NewTableStructByName("")

	selectStmt, ok := stmt.(*sqlparser.Select)
	if !ok {
		return nil, log.Error(errors.New("sql statement is only select supported"))
	}

	selectStmt.Limit = &sqlparser.Limit{
		Rowcount: sqlparser.NewIntVal([]byte("1")),
	}

	if selectStmt.From == nil || len(selectStmt.From) == 0 {
		return nil, log.Error(errors.New("Sql语句缺少From子句"))
	}

	if tableExpr, ok := selectStmt.From[0].(*sqlparser.AliasedTableExpr); ok {
		if tableName, ok := tableExpr.Expr.(*sqlparser.TableName); ok {
			newTs.SetName(tableName.Name.String())
		}
	}

	buffer := sqlparser.NewTrackedBuffer(nil)
	selectStmt.Format(buffer)
	queryStr := buffer.String()

	row, err := s.db.Query(queryStr)
	if err != nil {
		return nil, log.Error(err)
	}
	defer row.Close()

	columns, err := row.Columns()
	if err != nil {
		return nil, log.Error(err)
	}

	columnTypes, err := row.ColumnTypes()
	if err != nil {
		return nil, log.Error(err)
	}

	for i, column := range columns {
		colType := columnTypes[i]
		field, err := newFieldType(column, colType.DatabaseTypeName())
		if err != nil {
			return nil, err
		}
		err = newTs.AddFieldType(*field)
		if err != nil {
			return nil, err
		}
	}

	selectStmt.Limit = &sqlparser.Limit{
		Rowcount: sqlparser.NewIntVal([]byte("?")),
		Offset:   sqlparser.NewIntVal([]byte("?")),
	}

	buffer = sqlparser.NewTrackedBuffer(nil)
	selectStmt.Format(buffer)
	queryStr = buffer.String()

	s.stmt, err = s.db.Prepare(queryStr)
	if err != nil {
		return nil, log.Error(err)
	}

	fmt.Printf("stmt = %+v\n", reflect.TypeOf(stmt))
	fmt.Printf("stmt = %s\n", queryStr)
	return newTs, nil
}

func (s *MysqlSource) initTableStruct(ctx interfaces.IContext) error {
	var err error
	var newTs *datapack.TableStruct

	oldTs := ctx.GetTableStruct()
	if oldTs == nil {
		return log.Error(fmt.Errorf("TableStruct is nil"))
	}

	script := oldTs.Sql()
	if len(script) > 0 {
		newTs, err = s.getSqlStruct(oldTs)
		if err != nil {
			return log.Error(err)
		}
	} else {
		newTs, err = s.getSchemaStruct(oldTs)
		if err != nil {
			return log.Error(err)
		}
	}

	err = ctx.SetTableStruct(newTs)
	if err != nil {
		return err
	}

	return nil
}

func (s *MysqlSource) Open(ctx interfaces.IContext) error {
	var err error
	log.Debug("connecting to mysql source: ", s.cfg.Database)
	s.db, err = Connect(s.cfg)
	if err != nil {
		return err
	}
	err = s.initTableStruct(ctx)
	if err != nil {
		return err
	}

	s.offset = 0
	log.Debug("connected to mysql source: ", s.cfg.Database)
	return nil
}

func (s *MysqlSource) Close() error {
	if s.db != nil {
		err := s.db.Close()
		if err != nil {
			return err
		}
	}
	s.db = nil
	return nil
}

func (s *MysqlSource) ReadRow(ctx interfaces.IContext) error {
	var err error
	tableStruct := ctx.GetTableStruct()
	if s.db == nil || tableStruct == nil {
		return log.Error(errors.New("mysql source is closed"))
	}
	startTime := time.Now()
	rows, err := s.stmt.Query(s.offset, MAX_ROWSBUFFER)
	if err != nil {
		return err
	}

	fieldCount := tableStruct.Length()

	data := make([]interface{}, fieldCount)
	dataPtr := make([]interface{}, fieldCount)
	for i := range fieldCount {
		dataPtr[i] = &data[i]
	}

	readCount := 0
	for rows.Next() {
		err = rows.Scan(dataPtr...)
		if err != nil {
			return log.Error(err)
		}

		rowData, err := tableStruct.Serialize(data)
		if err != nil {
			return err
		}

		err = ctx.SendRowData(&rowData)
		if err != nil {
			return err
		}
		readCount++
	}
	s.offset += readCount

	if err = rows.Err(); err != nil {
		_ = log.Error(err)
	}

	if readCount < MAX_ROWSBUFFER {
		log.Debug("send end row")
		err = ctx.SendRowData(nil)
		if err != nil {
			return err
		}
		ctx.Stop()
	}

	endTime := time.Now()
	spendTime := endTime.Sub(startTime)
	if spendTime > time.Second {
		log.Debug(fmt.Sprintf("-- total rows: %d, read time: %fs", readCount, spendTime.Seconds()))
	}

	return nil
}

func CreateMysqlSource(params map[string]interface{}) (interfaces.IComponentFunction, error) {
	var err error
	src := MysqlSource{}
	src.cfg, err = parseConfig(params)
	if err != nil {
		return nil, err
	}
	return &src, nil
}
